package org.example.api.tableapi.udf;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/3/9
 */
public class ScalarFunc {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 基于数据流创建表
        Table table = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp ");
        // 4
        // 4.1 table api
        HashCode hashCode = new HashCode(13);
        tableEnv.registerFunction("hashCode", hashCode);
        Table res = table.select("id,ts,hashCode(id)");

        // 4.2 sql
        tableEnv.createTemporaryView("sensor", table);
        Table sqlRes = tableEnv.sqlQuery("select id, ts, hashCode(id) from sensor");

        tableEnv.toAppendStream(res, Row.class).print("res");
        tableEnv.toAppendStream(sqlRes, Row.class).print("sqlRes");

        env.execute();

    }

    /**
     * 实现自定义的标量函数
     */
    public static class HashCode extends ScalarFunction {
        private int factor = 13;

        public HashCode(int factor) {
            this.factor = factor;
        }

        public int eval(String str) {
            return str.hashCode() * factor;
        }

    }
}
